package net.lesscoding.producer;

import net.lesscoding.common.Const;
import net.lesscoding.utils.MessageListSplitter;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

import java.util.ArrayList;
import java.util.List;

/**
 * @author eleven
 * @date 2024/6/1 14:54
 * @apiNote 批量消息生产者
 */
public class BatchProducer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        producer.setNamesrvAddr(Const.TS_NAME_SRV_ADDR_OS1);
        /**
         * 指定要发送的消息最大大小，默认为4M
         * 不过仅修改这个属性是不行的，还需要同时修改broker加载的配置文件中的
         * maxMessageSize属性值
         */
        producer.setMaxMessageSize(4 * 1024 * 1024);
        producer.start();

        List<Message> msgList = new ArrayList<>();
        Message msg = null;
        for (int i = 0; i < 100; i++) {
            byte[] body = ("Hi," + i).getBytes();
            msg = new Message("BatchTopic", "BatchTag", body);
            msgList.add(msg);
        }

        MessageListSplitter splitter = new MessageListSplitter(msgList);
        while (splitter.hasNext()) {
            try {
                List<Message> listItem = splitter.next();
                producer.send(listItem);
                listItem.forEach(System.out::println);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.shutdown();
    }
}
